package com.qupeng.demo.kafka.kafkaspringbootproducer.producer;

import com.qupeng.demo.kafka.kafkaspringbootproducer.config.KafkaProperties;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class ProducerFactoryConfig<K, V> {

    @Autowired
    private KafkaProperties kafkaProperties;

    @Bean
    public Map<String, Object> producerConfigs() throws ClassNotFoundException, IOException {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServer());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, Class.forName(kafkaProperties.getKeySerializerInt()));
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, kafkaProperties.getSecurityProtocol());
        if (kafkaProperties.getSecurityProtocol().contains("SSL")) {
            String keystorePath = kafkaProperties.getSslKeystoreLocation().startsWith("classpath") ? kafkaProperties.getKeystore().getFile().getAbsolutePath() : kafkaProperties.getSslKeystoreLocation();
            props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keystorePath);
            props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, kafkaProperties.getSslKeystorePassword());
            if (StringUtils.hasText(kafkaProperties.getSslKeyPassword())) {
                props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, kafkaProperties.getSslKeyPassword());
            }
            props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, kafkaProperties.getSslKeystoreType());

            String truststorePath = kafkaProperties.getSslTruststoreLocation().startsWith("classpath") ? kafkaProperties.getTruststore().getFile().getAbsolutePath() : kafkaProperties.getSslTruststoreLocation();
            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststorePath);
            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, kafkaProperties.getSslTruststorePassword());
            props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, kafkaProperties.getSslTruststoreType());
        }
        if (StringUtils.hasText(kafkaProperties.getSaslMechanism())) {
            props.put(SaslConfigs.SASL_MECHANISM, kafkaProperties.getSaslMechanism());
            props.put(SaslConfigs.SASL_JAAS_CONFIG, kafkaProperties.getSaslJaasConfig());
        }
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "rest-api");
        switch (kafkaProperties.getSerializationFormat()) {
            case "json":
                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Class.forName(kafkaProperties.getValueSerializerJson()));
                break;
            case "avro":
                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Class.forName(kafkaProperties.getValueSerializerAvro()));
                props.put("schema.registry.url", kafkaProperties.getSchemaRegistryUrl());
                break;
            case "str":
            default:
                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Class.forName(kafkaProperties.getValueSerializerStr()));
        }
        props.put(JsonSerializer.TYPE_MAPPINGS, "product:com.qupeng.demo.kafka.kafkaspringbootproducer.model.Product");
        return props;
    }

    @Bean
    public ProducerFactory<K, V> producerFactory() throws ClassNotFoundException, IOException {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
}
